#ifndef 	_UDP_SOURCE_
#define		_UDP_SOURCE_

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <glib.h>

#include <queue>
#include <vector>
#include "udp_ts_protocol.h"

#define MAX_EPOLL_SIZE              256
#define MAX_REC_BUFFER 				1024*6
#define MAX_STREAM_NAME_LEN 		30
#define DEFULT_MAX_CONNECT_NUM		256
#define DEFULT_UDP_SERVER_PORT		1635
#define DEFULT_UDP_BUFFER_LEN		50

#define WRITE_TS_FILE

struct cmp
{
	inline bool operator ()(ts_dd * a,ts_dd * b)
	{
		uint32_t anum  = udpTsProtocolParser::getPkgSequence(a->data);
		uint32_t bnum  = udpTsProtocolParser::getPkgSequence(b->data);

		return anum > bnum;
	}
};

class udpClientChannel
{
public:
	udpClientChannel()
	{
		memset(m_stream_name,0,sizeof(m_stream_name));
		m_is_run = false;
		g_mutex_init(&m_que_mutex);
#ifdef WRITE_TS_FILE
		m_ts_file = NULL;
#endif
	}
	~udpClientChannel()
	{

		while(!m_queue.empty())
		{
			ts_dd* tsData = m_queue.top();
			if(tsData->data)
			{
				delete [] tsData->data;
			}
			delete tsData;
			m_queue.pop();
		}
		g_mutex_clear(&m_que_mutex);
#ifdef WRITE_TS_FILE
		if(m_ts_file)
		{
			fclose(m_ts_file);
		}
#endif
	}
	inline void udpClientFflushQueue()
	{
		while(!m_queue.empty())
		{
			ts_dd* tsData = m_queue.top();
			if(tsData->data)
			{
				delete [] tsData->data;
			}
			delete tsData;
			m_queue.pop();
		}
	}
	inline char* udpClientGetStreamName()
	{
		return m_stream_name;
	}
	inline void udpClientPushTsData(ts_dd* data)
	{
		g_mutex_lock(&m_que_mutex);
		m_queue.push(data);
		g_mutex_unlock(&m_que_mutex);
	}
	inline ts_dd* udpClientGetTsData()
	{
		g_mutex_lock(&m_que_mutex);
		if(m_queue.size() > DEFULT_UDP_BUFFER_LEN)
		{
			ts_dd *ts = m_queue.top();
			m_queue.pop();
			g_mutex_unlock(&m_que_mutex);
			return ts;
		}
		else
		{
			g_mutex_unlock(&m_que_mutex);
			return NULL;
		}
	}
	inline bool udpClientIsRun()
	{
		return m_is_run;
	}
	inline void udpClientSetState(bool state)
	{
		m_is_run = state;
	}

#ifdef WRITE_TS_FILE
public:
	FILE *m_ts_file;
#endif

private:
	GMutex m_que_mutex;
	bool m_is_run;
	char m_stream_name[21+MAX_STREAM_NAME_LEN];   //ip:port#stream    192.168.12.40:1635#treacher
	std::priority_queue<ts_dd*,std::vector<ts_dd*>,cmp > m_queue;
};

class udpSource
{
public:
	udpSource();
	~udpSource();
	int udpServerConfig(const char* config_file);
	int udpServerStart();
	int udpServerStop();
private:
	int udpServerInit();
	uint32_t udpServerHash(struct sockaddr_in *addr);
private:
	static void* listenThreadFunc(void* data);
	static void threadPoolFunc(void* data,void*user_data);
private:
	int						m_listener;
	int 					m_port;
	int						m_epfd;
	struct epoll_event 		m_ev;
	struct epoll_event 		m_events[MAX_EPOLL_SIZE];
	GThread *				m_listen_thread;
	bool					m_listen_thread_stop;
	GThreadPool *			m_thread_pool;
	int						m_thread_pool_size;
	int						m_max_connect_num;
	int						m_cur_connect_num;
	udpClientChannel *		m_client_list;
	int						m_client_timeout;
};

#endif
